package linelossHadoop;

import java.io.IOException;
import java.math.BigDecimal;
import java.math.RoundingMode;
import java.util.HashMap;
import java.util.Map.Entry;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.TextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

public class NodeLineLoss {

	public static class Map extends Mapper<Object, Text, Text, Text> {
		public void map(Object key, Text value, Context context)
				throws IOException, InterruptedException {
			String line = value.toString();
			if (line == null
					|| line.equals("")
					|| line.startsWith("name,totalnodelengthforroot,level,children,in,out"))
				return;
			String[] words = line.split(",");
			if (words == null)
				return;
			if (words[0].equals("0")) {
				context.write(new Text("totalIn"), new Text(words[5]));
				// System.out.println("nodes:"+words[1]);
				return;
			}
			if (words[3].length() == 0) {
				// NodeLineLoss.nodeMapIn.put(Integer.parseInt(words[0]),
				// Double.valueOf(words[5]));
				// actOut+=Double.valueOf(words[5]);
				context.write(new Text("actOut"), new Text(words[5]));
				return;
			} else {
				// System.out.println(words[0]);
				// NodeLineLoss.nodeMapIn.put(Integer.parseInt(words[0]),
				// Double.valueOf(words[4]));
				// context.write(new
				// Text((3-Integer.parseInt(words[2]))+":"+words[0]), new
				// Text(words[3]+":"+words[4]+":"+words[5]));
				//System.out.println(words[0] + " " + words[4] + ":" + words[5]);
				context.write(new Text((5 - Integer.parseInt(words[2])) + ":"
						+ words[0]), new Text(words[3] + ":" + words[5]));
				return;
				// children,in,out
			}
		}
	}

	public static class Reduce extends Reducer<Text, Text, Text, Text> {
		private MultipleOutputs<Text, Text> mos;

		@Override
		protected void setup(Context context) throws IOException,
				InterruptedException {
			mos = new MultipleOutputs<Text, Text>(context);
		}

		public void reduce(Text key, Iterable<Text> values, Context context)
				throws IOException, InterruptedException {
//			if (context.getConfiguration().get("rootIn") != null
//					&& context.getConfiguration().get("actOut") != null) {
//				double actOut = Double.valueOf(context.getConfiguration().get(
//						"rootIn"));
//				double totalIn = Double.valueOf(context.getConfiguration().get(
//						"totalIn"));
//				this.mos.write("lineloss",
//						new Text("ComprehensiveLineLossRate"),
//						new Text(Double.toString(1 - actOut / totalIn)));
////				context.getConfiguration().unset("rootIn");
////				context.getConfiguration().unset("totalIn");
//			}
			if (key.toString().equals("totalIn")) {
				System.out.println("here");
				double totalIn = 0;
				for (Text val : values) {
					totalIn += Double.valueOf(val.toString());
				}
				this.mos.write("lineloss", new Text("rootIn"),
						new Text(Double.toString(totalIn)));
//				context.getConfiguration().set("rootIn",
//						Double.toString(totalIn));
				return;
			} else if (key.toString().equals("actOut")) {
				double actOut = 0;
				for (Text val : values) {
					actOut += Double.valueOf(val.toString());
				}
				this.mos.write("lineloss", new Text("rootActConsume"),
						new Text(Double.toString(actOut)));
//				context.getConfiguration().set("actOut",
//						Double.toString(actOut));
				return;
			} else {
				String[] t = key.toString().split(":");
				for (Text val : values) {
					//System.out.println(val);
					BigDecimal out = new BigDecimal(0);
//					double out=0;
					// System.out.println(val.toString());
					for (String child : val.toString().split(":")[0].split(" ")) {
						if (child == null) {
							continue;
						} else {
							out=out.add(new BigDecimal(child));
//							out+=Double.parseDouble(child);
							//System.out.println(out+" "+new BigDecimal(child));
						}
					}
					if (val.toString().split(":").length == 2) {
						// System.out.println(Double.valueOf(val.toString().split(":")[2])+" "+out);
						BigDecimal actResult=new BigDecimal(val.toString().split(":")[1]);						
						actResult=actResult.subtract(out);
//						System.out.println(actResult);
						actResult= actResult.divide(new BigDecimal(val.toString().split(":")[1]),10,RoundingMode.HALF_DOWN);
						//String output=actResult.doubleValue();
//						double actResult = (Double.parseDouble(val.toString().split(":")[1]) - out)/Double.parseDouble(val.toString().split(":")[1]);
//						if (actResult.) {
//							System.out.println(Double.parseDouble(val.toString()
//									.split(":")[1]) + " " + out);
//						}
						String output=String.format("%.7f", actResult.doubleValue()*100);
						//String output = String.format("%.7f", actResult.doubleValue()*100);
						this.mos.write("lineloss", new Text(t[1]), new Text(
								output));
					} else {
						// System.out.println(key);
					}
				}
			}
		}

		@SuppressWarnings("unchecked")
		@Override
		protected void cleanup(
				org.apache.hadoop.mapreduce.Reducer.Context context)
				throws IOException, InterruptedException {
			super.cleanup(context);
			this.mos.close();
		}
	}

	@SuppressWarnings("deprecation")
	public void runSelf(String[] args) throws Exception {
		Configuration conf = new Configuration();
		conf.setInt("mapred.ifile.buffer.reset.size.mb", 256);
		Job job = new Job(conf, "NodeLineLoss");
		job.setJarByClass(NodeLineLoss.class);
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(Text.class);
		job.setMapperClass(Map.class);
		job.setNumReduceTasks(1);
		// job.setPartitionerClass(myPartition.class);
		MultipleOutputs.addNamedOutput(job, "lineloss", TextOutputFormat.class,
				Text.class, Text.class);
		job.setCombinerClass(Reduce.class);
		job.setReducerClass(Reduce.class);

		job.setInputFormatClass(TextInputFormat.class);
		// job.setOutputFormatClass(TextOutputFormat.class);

		FileInputFormat.addInputPath(job, new Path(args[0]));
		FileOutputFormat.setOutputPath(job, new Path(args[1]));

		job.waitForCompletion(true);
	}

	public static void main(String[] args) throws Exception {
		long timeTestStart = System.currentTimeMillis();// 记录开始时间
		NodeLineLoss myCalculate = new NodeLineLoss();
		myCalculate.runSelf(args);
		long timeTestEnd = System.currentTimeMillis();// 记录结束时间
		System.out.println(String.format("运行时间是%d毫秒", timeTestEnd
				- timeTestStart));
	}

}
